package com.niit.service

import com.niit.bean.AnswerWithRecommendations
import com.niit.dao.EDUBatchDao
import com.niit.utils.SparkUtils
import org.apache.spark.sql.{DataFrame, Dataset, Row}

/**
 * Date:2024/5/30
 * Author：Ys
 * Description:
 */
class EDUBatchService {
  private val spark = SparkUtils.takeSpark()

  import spark.implicits._
  import org.apache.spark.sql.functions._

  def dataAnalysis(): Unit = {
    //1.读取数据库 edu表的数据
    val eduDao = new EDUBatchDao
    val answerData: Dataset[AnswerWithRecommendations] = eduDao.getEDUData()

    hotSubjectCountTop50(answerData)
    hotSubjectRecommendTop20(answerData)
  }
  //需求一：找到Top50热点题对应的科目，然后再这些科目中，分别包含热点题的数据

  def hotSubjectCountTop50(answerData: Dataset[AnswerWithRecommendations]): Unit = {
    //1.统计前50道热点题---》 在数据库中，即使相同的题目，也是分布不在的行中
                              //根据题目id进行分组，分组后统计每个题目出现的次数，并进行降序取前50条数据
    val hotTop50: Dataset[Row] = answerData.groupBy("question_id").agg(count("*") as "hot").orderBy('hot.desc).limit(50)
    // 结果： question_id         count
    //2. 将Top50结果 和 answerData进行关联，得到热点题对应科目， 对重复的题目进行去重
    val joinDF: DataFrame = hotTop50.join(answerData.dropDuplicates("question_id"), "question_id")

    //3.按学科分组聚合统计每个学科包含热点题数量
    val res: Dataset[Row] = joinDF.groupBy("subject_id")
      .agg(count("*") as "hotCount")
      .orderBy('hotCount.desc)

    res.show()
  }

  //需求二：各个科目推荐题分析
  /*
    找到前20热点题对应的推荐题目，然后找到推荐题对应的科目，统计每个科目分别包含推荐题的数量
   */
  def hotSubjectRecommendTop20(answerData: Dataset[AnswerWithRecommendations]): Unit = {
    //1.统计前20道热点题，根据数量进行降序
    val hotTop20: Dataset[Row] = answerData.groupBy("question_id").agg(count("*") as "hot").orderBy('hot.desc).limit(20)

    //2.将hotTop20 和 answerData 进行关联，得到热点题的推荐列表
    val ridsDF: DataFrame = hotTop20.join(answerData, "question_id").select("recommendations")
    // 结果 question_id    recommendations
    //3.将获得的题目对应的推荐列表（ridsDF） 转换成数据  对推荐列表中的题目按照","进行切割，目的根据题目找到对应的科目
    val ridDFS: Dataset[Row] = ridsDF.select(explode(split('recommendations, ",") ) as "question_id").dropDuplicates("question_id")
    //4.将ridDFS 和 answerData进行管理
    val ridAndSid: DataFrame = ridDFS.join(answerData.dropDuplicates("question_id"), "question_id")
    //5.统计各个科目包含的推荐题数据，并降序排序
    ridAndSid.groupBy("subject_id").agg(count("*") as "rcount").orderBy('rcount.desc).show()
  }

}
